package com.lyz.remote.impl;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.impl.conn.PoolingClientConnectionManager;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.log4j.Logger;

import com.lyz.config.ConfigFactory;
import com.lyz.config.IConfig;
import com.lyz.core.impl.Context;
import com.lyz.remote.IMsgHandler;
import com.lyz.util.ClassUtil;

/**
 * 默认http协议分发器，主要用于读取一份数据，向多个节点分发
 * @author luyongzhao
 *
 */
public class DefaultHttpGroupMsgHandler implements IMsgHandler{
	
	private static final Logger logger = Logger.getLogger(DefaultHttpGroupMsgHandler.class);
	/*实际发送数据的任务类的全称*/
	private List<String> taskNodeNames = null;
	
	private List<String> urls = null;
	
	private DefaultHttpClient httpclient = null;
	
	private ThreadPoolExecutor pool = null;
	
	private IConfig config = null;
	
	public DefaultHttpGroupMsgHandler(){
		PoolingClientConnectionManager manager = new PoolingClientConnectionManager();
		manager.setMaxTotal(8);
		httpclient = new DefaultHttpClient(manager);
		config = ConfigFactory.getConfig();
		pool = new ThreadPoolExecutor(2, 20, 10, TimeUnit.SECONDS, 
    			new LinkedBlockingQueue<Runnable>(10),new ThreadPoolExecutor.CallerRunsPolicy());
		logger.info("init http thread pool...!");
	}

	public IMsgHandler initHandler(String tag, Context context) {
		/*设置请求超时时间*/
		httpclient.getParams().setIntParameter(CoreConnectionPNames.SO_TIMEOUT, config.getSoTimeout(tag));
		taskNodeNames = config.getTaskNodeNames(tag);
		urls = config.getUris(tag);
		return this;
	}

	public boolean sendMsg(String msg) {
		
		if(msg==null || "".equals(msg.trim())){
			return true;
		}
		/*遍历所有的发送器，发送数据*/
		for(int i=0; i<taskNodeNames.size(); i++){
			if(taskNodeNames.get(i)==null || "".equals(taskNodeNames.get(i).trim())){
				continue;
			}
			/*获取执行实例*/
			AbstractHttpSender sender = (AbstractHttpSender)ClassUtil.newInstance(taskNodeNames.get(i));
			if(sender != null){
				/*配置文件中的sender类必须与url一一对应*/
				sender.init(httpclient, urls.get(i), msg);
				pool.execute(sender);
			}
		}
				
		return true;	
	}

	public boolean sendMsg(byte[] msg) {
		logger.warn("sendMsg(byte[] msg) is not implemented!");
		return false;
	}
	
	
	
}
